package com.stone.boot.pipeline.app.service.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
import com.stone.boot.pipeline.app.service.PipelineNodeService;
import com.stone.boot.pipeline.app.service.PipelineRedisService;
import com.stone.boot.pipeline.app.service.PipelineService;
import com.stone.boot.pipeline.domain.entity.Pipeline;
import com.stone.boot.pipeline.domain.entity.PipelineNode;
import com.stone.boot.pipeline.domain.repository.PipelineNodeRepository;
import com.stone.boot.pipeline.domain.repository.PipelineRepository;
import com.stone.boot.pipeline.domain.vo.PipelineNodeVO;
import com.stone.boot.pipeline.domain.vo.PipelineVO;
import com.stone.boot.pipeline.infra.constant.StonePipelineConstant;
import com.stone.starter.core.constant.BaseConstants;
import com.stone.starter.core.exception.CacheModifyException;
import com.stone.starter.core.exception.CommonException;
import com.stone.starter.core.helper.TransactionalHelper;
import com.stone.starter.core.util.list.ListUtil;
import com.stone.starter.redis.client.RedisCacheClient;
import lombok.RequiredArgsConstructor;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.hzero.mybatis.domian.Condition;
import org.hzero.mybatis.util.Sqls;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cglib.beans.BeanCopier;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;

import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * 流程器service实现
 *
 * @author Mr_wenpan@163.com 2021/8/5 10:18 下午
 */
@Service
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class PipelineServiceImpl implements PipelineService {

    private final PipelineRepository pipelineRepository;

    private final PipelineNodeRepository pipelineNodeRepository;

    private final TransactionalHelper transactionalHelper;

    private final PipelineNodeService pipelineNodeService;

    private final PipelineRedisService pipelineRedisService;

    private final RedisCacheClient redisCacheClient;

    @Override
    public int batchMerge(List<Pipeline> pipelines) {
        int existNum = 0;
        for (Pipeline pipeline : pipelines) {
            // 非空字段校验
            pipeline.validate();
            // 编码是否重复
            if (pipeline.exist(pipelineRepository)) {
                existNum++;
                continue;
            }
            transactionalHelper.transactionOperation(() -> {
                if (pipeline.getId() != null) {
                    pipelineRepository.updateByPrimaryKey(pipeline);
                    boolean isSuccess = pipelineRedisService.saveRedisPipelineNodeConf(pipeline.getId());
                    // 如果lua脚本更新Redis执行失败了，则抛出异常回滚数据保证数据一致性
                    if (!isSuccess) {
                        throw new CacheModifyException("缓存更新异常，流水线id为 [%s]", pipeline.getId());
                    }
                } else {
                    pipelineRepository.insertSelective(pipeline);
                    pipelineRedisService.createRedisPipelineNodeConf(pipeline.getId());
                }
            });
        }
        return existNum;
    }

    @Override
    public void batchCreateAndUpdate(List<Pipeline> pipelines) {
        if (CollectionUtils.isEmpty(pipelines)) {
            return;
        }
        ImmutablePair<List<Pipeline>, List<Pipeline>> pair =
                ListUtil.splitList(pipelines, data -> Objects.isNull(data.getId()));
        transactionalHelper.transactionOperation(() -> {
            // 批量新增
            batchCreateCommon(pair.left);
            // 批量修改
            batchUpdateCommon(pair.right);
        });
    }

    @Override
    public PipelineVO getPipelineDTO(Long id) {
        Pipeline pipeline = pipelineRepository.selectByPrimaryKey(id);
        List<PipelineNode> pipelineNodes = pipelineNodeRepository.select(PipelineNode.FIELD_PIPELINE_ID, id);
        if (null == pipeline) {
            throw new CommonException("未找到该流程器");
        }
        return buildPipelineDTO(pipeline, pipelineNodes);
    }

    @Override
    public void batchUpdate(List<Pipeline> pipelines) {
        if (CollectionUtils.isEmpty(pipelines)) {
            return;
        }
        batchUpdateCommon(pipelines);
    }

    @Override
    public void batchInsert(List<Pipeline> pipelines) {
        if (CollectionUtils.isEmpty(pipelines)) {
            return;
        }
        batchCreateCommon(pipelines);
    }

    @Override
    public void savePipelineDTOS(List<MultipartFile> multipartFiles) {
        // 使用ObjectMapper来解析yml文件
        ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
        // 使用BeanCopier来拷贝数据，BeanCopier性能上比Spring的BeanUtils，Apache的BeanUtils和PropertyUtils要好很多
        BeanCopier pipelineCopier = BeanCopier.create(PipelineVO.class, Pipeline.class, false);
        BeanCopier pipelineNodeCopier = BeanCopier.create(PipelineNodeVO.class, PipelineNode.class, false);

        // 这里是垃圾代码，需要重写
        transactionalHelper.transactionOperation(() -> {
            for (MultipartFile file : multipartFiles) {
                try {
                    // 直接读取流，转化为对象，不存储文件
                    PipelineVO pipelineDTO = mapper.readValue(file.getInputStream(), PipelineVO.class);
                    Pipeline pipeline = new Pipeline();
                    pipelineCopier.copy(pipelineDTO, pipeline, null);

                    // 插入流程器
                    batchInsert(Collections.singletonList(pipeline));
                    Map<String, PipelineNodeVO> pipelineDTONodes = pipelineDTO.getPipelineNodes();
                    List<PipelineNode> pipelineNodes = new ArrayList<>(pipelineDTONodes.size());

                    // 流程节点
                    pipelineDTONodes.forEach((key, pipelineNodeDTO) -> {
                        PipelineNode pipelineNode = new PipelineNode();
                        pipelineNodeCopier.copy(pipelineNodeDTO, pipelineNode, null);
                        pipelineNode.setPipelineId(pipeline.getId());
                        pipelineNodes.add(pipelineNode);
                    });

                    pipelineNodeService.batchCreate(pipelineNodes);

                } catch (IOException e) {
                    throw new CommonException("上传失败，确认yaml格式是否正确" + e.getMessage());
                }
            }
        });

    }

    @Override
    public void batchRemove(List<Pipeline> pipelines) {
        if (CollectionUtils.isEmpty(pipelines)) {
            return;
        }
        // 保证删除数据一致性
        transactionalHelper.transactionOperation(() -> {
            pipelineRepository.batchDeleteByPrimaryKey(pipelines);
            List<String> codeList = pipelines.stream().map(Pipeline::getCode).collect(Collectors.toList());
            Set<String> keys = new HashSet<>();
            for (String code : codeList) {
                keys.add(buildInterfaceConfKey(code));
            }
            // 批量删除keys，一次RTT
            redisCacheClient.delete(keys);
        });
    }

    private PipelineVO buildPipelineDTO(Pipeline pipeline, List<PipelineNode> pipelineNodes) {
        PipelineVO pipelineDTO = new PipelineVO();
        pipelineDTO.setStartAction(pipeline.getStartAction());
        pipelineDTO.setEndAction(pipeline.getEndAction());
        // pipelineDTO.setCode(pipeline.getCode());
        pipelineDTO.setPipelineCode(pipeline.getCode());
        pipelineDTO.setIsEnable(pipeline.getIsEnable());
        for (PipelineNode pipelineNode : pipelineNodes) {
            PipelineNodeVO pipelineNodeDTO = new PipelineNodeVO();
            pipelineNodeDTO.setCurAction(pipelineNode.getCurAction());
            pipelineNodeDTO.setNextAction(pipelineNode.getNextAction());
            pipelineNodeDTO.setCurDescription(pipelineNode.getCurDescription());
            pipelineNodeDTO.setStrategyType(pipelineNode.getStrategyType());
            pipelineDTO.getPipelineNodes().put(pipelineNodeDTO.uniqueKey(), pipelineNodeDTO);
        }

        return pipelineDTO;
    }

    /**
     * 批量创建流程器公用方法
     *
     * @param createList 批量创建的流程器集合
     * @author Mr_wenpan@163.com 2021/8/6 3:53 下午
     */
    private void batchCreateCommon(List<Pipeline> createList) {
        if (CollectionUtils.isEmpty(createList)) {
            return;
        }
        // 必输参数校验
        createList.forEach(Pipeline::validate);
        List<String> pipelineCoeList = createList.stream().map(Pipeline::getCode).collect(Collectors.toList());
        List<Pipeline> pipelineList = pipelineRepository.selectByCondition(Condition.builder(Pipeline.class)
                .andWhere(Sqls.custom().andIn(Pipeline.FIELD_CODE, pipelineCoeList)).build());
        Map<String, Pipeline> existedMap = pipelineList.stream()
                .collect(Collectors.toMap(e -> e.getCode() + e.getIsEnable(), e -> e));
        StringJoiner joiner = new StringJoiner(BaseConstants.Symbol.COMMA);

        if (MapUtils.isNotEmpty(existedMap)) {
            // 检查要新增的数据是否已经在数据库存在，若存在则拼接错误提示
            for (Pipeline pipeline : createList) {
                if (Objects.nonNull(existedMap.get(pipeline.getCode() + pipeline.getIsEnable()))) {
                    joiner.add(pipeline.getCode());
                }
            }
            if (StringUtils.isNotBlank(joiner.toString())) {
                throw new CommonException("流程器重复，如下流程器已经在数据库存在，请修正：{}", joiner.toString());
            }
        }

        transactionalHelper.transactionOperation(() -> {
            pipelineRepository.batchInsertSelective(createList);
            createList.forEach(e -> pipelineRedisService.createRedisPipelineNodeConf(e.getId()));
        });
    }

    /**
     * 批量更新流程器接口公用方法提取
     *
     * @param updateList 待更新的数据
     * @author Mr_wenpan@163.com 2021/8/6 3:58 下午
     */
    private void batchUpdateCommon(List<Pipeline> updateList) {
        if (CollectionUtils.isEmpty(updateList)) {
            return;
        }
        // 必输参数校验
        updateList.forEach(Pipeline::validate);
        transactionalHelper.transactionOperation(() -> {
            pipelineRepository.batchUpdateByPrimaryKeySelective(updateList);
            // 利用lua脚本去更新Redis缓存数据，一旦更新失败回滚数据，保证一致性
            updateList.forEach(e -> {
                if (!pipelineRedisService.saveRedisPipelineNodeConf(e.getId())) {
                    throw new CacheModifyException("更新数据到缓存中发生异常，请重新操作.");
                }
            });
        });
    }

    /**
     * @param pipelineCode 流程器配置代码
     * @return redis cache key
     */
    public String buildInterfaceConfKey(String pipelineCode) {
        return String.format(StonePipelineConstant.Redis.PIPELINE_KEY, pipelineCode);
    }

}